package top.jolyoulu.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @Author: JolyouLu
 * @Date: 2024/5/19 15:07
 * @Description
 */
object Spark01_SparkStreaming_State_Window1 {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    val ssc = new StreamingContext(sparkConf, Seconds(3))
    ssc.checkpoint("cp")
    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.88.100", 9999)
    val wordToOne: DStream[(String, Int)] = lines.map((_, 1))
    //窗口范围比较大滑动浮动比较小时，可以使用增加数据和删除数据方式优化重复计算
    val windowDS: DStream[(String, Int)] = wordToOne.reduceByKeyAndWindow(
      (x:Int,y:Int) => {x + y}, //加上新进入窗口的批次数据
      (x:Int,y:Int) => {x - y}, //移除离开窗口的老批次数据
      Seconds(9), //窗口时长
      Seconds(3)  //滑动步长
    )
    val wordToCount: DStream[(String, Int)] = windowDS.reduceByKey(_ + _)
    wordToCount.print()

    ssc.start()
    ssc.awaitTermination()
  }
}
